package com.lagou.cl.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class WorkClient {

	private Configuration configuration = null;

	private Connection connection = null;

	@Before
	public void init() throws IOException {

		configuration = HBaseConfiguration.create();
		// 获取hbase的集群
		configuration.set("hbase.zookeeper.quorum", "linux111,linux112,linux113");
		configuration.set("hbase.zookeeper.property.clientPort", "2181");

		// 配置参数
		configuration.set("zookeeper.znode.parent", "/hbase");

		// 获取连接
		connection = ConnectionFactory.createConnection(configuration);
	}


	@Test
	public void createUserInfoTable() throws IOException {
		// 获取hbaseadmin对象
		HBaseAdmin hBaseAdmin = (HBaseAdmin) connection.getAdmin();
		// 创建表
		HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf("user_relations"));
		// 指定列簇
		descriptor.addFamily(new HColumnDescriptor("friends"));
		hBaseAdmin.createTable(descriptor);
	}

	@Test
	public void addFriends() throws IOException {
		// 获取表对象
		Table resume = connection.getTable(TableName.valueOf("user_relations"));
		String currentUserId = "1";
		String[] friendsOfCurrentUser = {"2", "3", "4"};
		List<Put> puts = new ArrayList<>();
		// 创建put对象，指定rowKey
		Put put = new Put(Bytes.toBytes(currentUserId));
		for (String fUid : friendsOfCurrentUser) {
			// 添加列
			put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(fUid), Bytes.toBytes(fUid));
			puts.add(put);
		}

		if (!puts.isEmpty()) {
			resume.put(puts);
		}
		resume.close();

	}

	@Test
	public void addFriends1() throws IOException {
		// 获取表对象
		Table resume = connection.getTable(TableName.valueOf("user_relations"));
		String currentUserId = "1";
		String[] friendsOfCurrentUser = {"2", "3", "4"};
		List<Put> cputs = new ArrayList<>();
		for (String cUid : friendsOfCurrentUser) {
			// 创建put对象，指定rowKey
			Put cPut = new Put(Bytes.toBytes(cUid));
			cPut.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(currentUserId), Bytes.toBytes(currentUserId));
			cputs.add(cPut);
		}
		resume.put(cputs);
		resume.close();
	}


	@Test
	public void deleteDataOfTable() throws IOException {

		// 获取表对象
		Table resume = connection.getTable(TableName.valueOf("user_relations"));
		// 创建delete对象，指定rowKey
		Delete delete = new Delete(Bytes.toBytes("1"));
		delete.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("3"));
		resume.delete(delete);
		resume.close();
	}

	@Test
	public void queryDataOfTable() throws IOException {
		// 获取表对象
		Table resume = connection.getTable(TableName.valueOf("user_relations"));
		// 获取get对象
		Get get = new Get(Bytes.toBytes("1"));
		// 指定查询的列簇
		get.addFamily(Bytes.toBytes("friends"));
		Result result = resume.get(get);
		dealWithData(result);
	}

	private void dealWithData(Result result) {
		List<Cell> cells = result.listCells();
		if (!cells.isEmpty()) {
			for (Cell cell : cells) {
				String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
				String columnsFamily = Bytes.toString(CellUtil.cloneFamily(cell));
				String columnQualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
				String columnsValue = Bytes.toString(CellUtil.cloneValue(cell));
				System.out.println(rowKey + "\t" + columnsFamily + ":" + columnQualifier + "\t" + columnsValue);
			}
		}
	}
}
